"""
股票分析工作流 - LangGraph并行执行架构

基于LangGraph实现完整的股票分析流程，包含技术、消息、估值、财务、资金五大维度分析。
使用StateGraph定义的并行边，支持维度选择和自定义事件流。
"""

import logging
import json
from typing import TypedDict, Annotated, AsyncIterator, Dict, Any, List, Optional, Literal
from datetime import datetime
import operator

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.config import get_stream_writer
from langchain_core.runnables import RunnableConfig

from agents.stock_technical_analysis_agent import stock_technical_analysis_agent
from agents.stock_news_analysis_agent import stock_news_analysis_agent
from agents.stock_valuation_analysis_agent import stock_valuation_analysis_agent
from agents.stock_fundamental_analysis_agent import stock_fundamental_analysis_agent
from agents.stock_capital_flow_analysis_agent import stock_capital_flow_analysis_agent
from agents.stock_investment_summary_agent import stock_investment_summary_agent


class StockAnalysisState(TypedDict):
    """股票分析状态 - 支持JSON数据格式"""
    # 输入
    stock_code: str
    user_query: str

    # ✅ 新增：分析维度列表（支持部分维度分析）
    analysis_dimensions: List[str]  # ["technical", "news", "valuation", "fundamental", "capital"]

    # ✅ 分析结果改为单个字符串（每个维度只有一个分析结果，避免累积）
    technical_analysis: str
    news_analysis: str
    valuation_analysis: str
    fundamental_analysis: str
    capital_flow_analysis: str

    # 最终总结
    investment_summary: str

    # ✅ 新增：输出格式控制
    output_format: Literal["text", "json"]  # 输出格式控制

    # ✅ 新增：JSON数据字段
    technical_analysis_json: Optional[dict]
    news_analysis_json: Optional[dict]
    valuation_analysis_json: Optional[dict]
    fundamental_analysis_json: Optional[dict]
    capital_flow_analysis_json: Optional[dict]
    investment_summary_json: Optional[dict]

    # 🆕 图表数据字段
    technical_charts: Optional[dict]      # 技术分析图表数据
    fundamental_charts: Optional[dict]    # 财务分析图表数据
    investment_charts: Optional[dict]     # 投资总结图表数据

    # 元数据
    errors: Annotated[List[str], operator.add]
    completed_steps: Annotated[List[str], operator.add]


class StockAnalysisWorkflow:
    """股票分析工作流 - 参考债券分析的优秀架构"""

    def __init__(self):
        """初始化工作流"""
        self.logger = logging.getLogger("workflows.stock_analysis")

        # 初始化专业分析智能体
        self.technical_agent = stock_technical_analysis_agent
        self.news_agent = stock_news_analysis_agent
        self.valuation_agent = stock_valuation_analysis_agent
        self.fundamental_agent = stock_fundamental_analysis_agent
        self.capital_agent = stock_capital_flow_analysis_agent
        self.summary_agent = stock_investment_summary_agent

        # ✅ 移除检查点保存器，避免状态累积
        # 每次分析都是独立的，不会复用之前的状态
        self.checkpointer = None

        # 构建工作流图
        self.app = self._build_workflow()

    def _build_workflow(self):
        """构建工作流状态图（支持维度选择的条件路由）"""
        # 创建状态图
        workflow = StateGraph(StockAnalysisState)

        # 添加所有节点
        workflow.add_node("technical_analysis", self._technical_analysis_node)
        workflow.add_node("news_analysis", self._news_analysis_node)
        workflow.add_node("valuation_analysis", self._valuation_analysis_node)
        workflow.add_node("fundamental_analysis", self._fundamental_analysis_node)
        workflow.add_node("capital_flow_analysis", self._capital_flow_analysis_node)
        workflow.add_node("investment_summary", self._investment_summary_node)

        # ✅ 添加dispatch节点（支持维度选择）
        workflow.add_node("dispatch", self._dispatch_node)

        # 设置入口点
        workflow.set_entry_point("dispatch")

        # ✅ 从dispatch节点使用条件边路由到选中的分析节点
        workflow.add_conditional_edges(
            "dispatch",
            self._route_from_dispatch,
            ["technical_analysis", "news_analysis", "valuation_analysis",
             "fundamental_analysis", "capital_flow_analysis"]
        )

        # ✅ 所有分析完成后汇聚到总结节点（条件边，确保只在选中的分析完成后执行总结）
        workflow.add_edge("technical_analysis", "investment_summary")
        workflow.add_edge("news_analysis", "investment_summary")
        workflow.add_edge("valuation_analysis", "investment_summary")
        workflow.add_edge("fundamental_analysis", "investment_summary")
        workflow.add_edge("capital_flow_analysis", "investment_summary")

        # 总结后结束
        workflow.add_edge("investment_summary", END)

        # 编译工作流（不使用checkpointer，确保每次分析独立）
        return workflow.compile()

    def _dispatch_node(self, state: StockAnalysisState) -> Dict[str, Any]:
        """
        分发节点：根据analysis_dimensions决定要执行的分析节点

        维度映射：
        - "technical": technical_analysis (技术分析)
        - "news": news_analysis (消息分析)
        - "valuation": valuation_analysis (估值分析)
        - "fundamental": fundamental_analysis (财务分析)
        - "capital": capital_flow_analysis (资金分析)
        """
        dimensions = state.get("analysis_dimensions", [])

        # 默认全量分析（如果未指定或维度>=4个，与债券分析保持一致）
        if not dimensions or len(dimensions) >= 4:
            dimensions = ["technical", "news", "valuation", "fundamental", "capital"]

        # 发送推理事件
        writer = get_stream_writer()
        writer({
            "type": "reasoning",
            "node": "dispatch",
            "content": f"📋 分析维度: {', '.join(dimensions)} ({len(dimensions)}个)",
            "timestamp": datetime.now().isoformat()
        })

        return {"analysis_dimensions": dimensions}

    def _route_from_dispatch(self, state: StockAnalysisState) -> List[str]:
        """
        路由函数：返回要执行的节点列表

        Returns:
            节点名称列表，LangGraph会并行执行这些节点
        """
        dimensions = state.get("analysis_dimensions", [])

        # 维度到节点的映射
        dimension_map = {
            "technical": "technical_analysis",
            "news": "news_analysis",
            "valuation": "valuation_analysis",
            "fundamental": "fundamental_analysis",
            "capital": "capital_flow_analysis"
        }

        # 根据维度返回要执行的节点
        nodes = [dimension_map[dim] for dim in dimensions if dim in dimension_map]

        # 至少执行一个节点（默认技术分析）
        if not nodes:
            nodes = ["technical_analysis"]

        self.logger.info(f"路由到节点: {nodes}")
        return nodes

    async def _technical_analysis_node(self, state: StockAnalysisState) -> Dict[str, Any]:
        """技术分析节点 - 支持JSON格式输出"""
        try:
            self.logger.info(f"开始技术分析: {state['stock_code']}")

            # ✅ 获取stream writer
            writer = get_stream_writer()

            # 发送节点开始事件
            writer({
                "type": "step_start",
                "node": "technical_analysis",
                "step_name": "📈 技术分析",
                "timestamp": datetime.now().isoformat()
            })

            results = []
            json_result = None

            # ✅ 根据输出格式调用智能体
            if state.get("output_format") == "json":
                # JSON格式输出 - 使用generate_technical_signals方法
                result = await self.technical_agent._generate_technical_signals(
                    stock_code=state['stock_code'],
                    output_format="json"
                )
                # 解析JSON结果
                try:
                    json_result = json.loads(result) if isinstance(result, str) else result
                    # 同时保持文本格式（向后兼容）
                    text_result = json.dumps(json_result, ensure_ascii=False, indent=2)
                    results.append(text_result)
                except Exception as json_error:
                    self.logger.warning(f"JSON解析失败，使用文本格式: {json_error}")
                    results.append(result)
            else:
                # 原有文本处理逻辑
                async for event in self.technical_agent.stream_analyze(state['stock_code']):
                    event_type = event.get("type")

                    if event_type == "reasoning":
                        # ✅ 转发所有推理内容，不再过滤
                        content = event.get("content", "")
                        if content:
                            writer({
                                "type": "reasoning",
                                "node": "technical_analysis",
                                "content": content,
                                "timestamp": datetime.now().isoformat()
                            })

                    elif event_type == "final_answer":
                        # 只保存结果，不发送到UI
                        results.append(event.get("content", ""))

            # 🆕 生成技术分析图表数据（只获取数据，不渲染）
            chart_data = {}
            try:
                self.logger.info(f"开始生成技术分析图表数据: {state['stock_code']}")
                chart_result = await self.technical_agent.generate_technical_charts(state['stock_code'])
                if chart_result.get("success"):
                    chart_data = chart_result
                    self.logger.info(f"✅ 技术分析图表数据生成成功")
                else:
                    self.logger.warning(f"⚠️ 技术分析图表数据生成失败: {chart_result.get('error')}")
            except Exception as chart_error:
                self.logger.error(f"❌ 生成技术分析图表数据异常: {chart_error}")

            # 发送节点结束事件
            writer({
                "type": "step_end",
                "node": "technical_analysis",
                "step_name": "📈 技术分析",
                "timestamp": datetime.now().isoformat()
            })

            return {
                "technical_analysis": "\n".join(results),  # ✅ 拼接为单个字符串
                "technical_analysis_json": json_result,    # ✅ 存储JSON数据
                "technical_charts": chart_data,            # 🆕 存储图表数据
                "completed_steps": ["technical_analysis"]
            }
        except Exception as e:
            self.logger.error(f"技术分析失败: {e}")
            writer = get_stream_writer()
            writer({
                "type": "error",
                "node": "technical_analysis",
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            })
            return {
                "technical_analysis": f"技术分析失败: {str(e)}",
                "technical_analysis_json": None,
                "errors": [f"技术分析失败: {str(e)}"],
                "completed_steps": ["technical_analysis"]
            }

    async def _news_analysis_node(self, state: StockAnalysisState) -> Dict[str, Any]:
        """消息分析节点 - 支持JSON格式输出"""
        try:
            self.logger.info(f"开始消息分析: {state['stock_code']}")

            # ✅ 获取stream writer
            writer = get_stream_writer()

            # 发送节点开始事件
            writer({
                "type": "step_start",
                "node": "news_analysis",
                "step_name": "📰 消息分析",
                "timestamp": datetime.now().isoformat()
            })

            results = []
            json_result = None

            # ✅ 根据输出格式调用智能体
            if state.get("output_format") == "json":
                # JSON格式输出 - 使用_generate_news_signals方法
                result = await self.news_agent._generate_news_signals(
                    stock_code=state['stock_code'],
                    output_format="json"
                )
                # 解析JSON结果
                try:
                    json_result = json.loads(result) if isinstance(result, str) else result
                    # 同时保持文本格式（向后兼容）
                    text_result = json.dumps(json_result, ensure_ascii=False, indent=2)
                    results.append(text_result)
                except Exception as json_error:
                    self.logger.warning(f"JSON解析失败，使用文本格式: {json_error}")
                    results.append(result)
            else:
                # 原有文本处理逻辑
                async for event in self.news_agent.stream_analyze(state['stock_code']):
                    event_type = event.get("type")

                    if event_type == "reasoning":
                        # ✅ 转发所有推理内容，不再过滤
                        content = event.get("content", "")
                        if content:
                            writer({
                                "type": "reasoning",
                                "node": "news_analysis",
                                "content": content,
                                "timestamp": datetime.now().isoformat()
                            })

                    elif event_type == "final_answer":
                        # 只保存结果，不发送到UI
                        results.append(event.get("content", ""))

            # 发送节点结束事件
            writer({
                "type": "step_end",
                "node": "news_analysis",
                "step_name": "📰 消息分析",
                "timestamp": datetime.now().isoformat()
            })

            return {
                "news_analysis": "\n".join(results),  # ✅ 拼接为单个字符串
                "news_analysis_json": json_result,    # ✅ 存储JSON数据
                "completed_steps": ["news_analysis"]
            }
        except Exception as e:
            self.logger.error(f"消息分析失败: {e}")
            writer = get_stream_writer()
            writer({
                "type": "error",
                "node": "news_analysis",
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            })
            return {
                "news_analysis": f"消息分析失败: {str(e)}",
                "news_analysis_json": None,
                "errors": [f"消息分析失败: {str(e)}"],
                "completed_steps": ["news_analysis"]
            }

    async def _valuation_analysis_node(self, state: StockAnalysisState) -> Dict[str, Any]:
        """估值分析节点 - 支持JSON格式输出"""
        try:
            self.logger.info(f"开始估值分析: {state['stock_code']}")

            # ✅ 获取stream writer
            writer = get_stream_writer()

            # 发送节点开始事件
            writer({
                "type": "step_start",
                "node": "valuation_analysis",
                "step_name": "💰 估值分析",
                "timestamp": datetime.now().isoformat()
            })

            results = []
            json_result = None

            # ✅ 根据输出格式调用智能体
            if state.get("output_format") == "json":
                # JSON格式输出 - 使用_generate_valuation_signals方法
                result = await self.valuation_agent._generate_valuation_signals(
                    stock_code=state['stock_code'],
                    output_format="json"
                )
                # 解析JSON结果
                try:
                    json_result = json.loads(result) if isinstance(result, str) else result
                    # 同时保持文本格式（向后兼容）
                    text_result = json.dumps(json_result, ensure_ascii=False, indent=2)
                    results.append(text_result)
                except Exception as json_error:
                    self.logger.warning(f"JSON解析失败，使用文本格式: {json_error}")
                    results.append(result)
            else:
                # 原有文本处理逻辑
                async for event in self.valuation_agent.stream_analyze(state['stock_code']):
                    event_type = event.get("type")

                    if event_type == "reasoning":
                        # ✅ 转发所有推理内容，不再过滤
                        content = event.get("content", "")
                        if content:
                            writer({
                                "type": "reasoning",
                                "node": "valuation_analysis",
                                "content": content,
                                "timestamp": datetime.now().isoformat()
                            })

                    elif event_type == "final_answer":
                        # 只保存结果，不发送到UI
                        results.append(event.get("content", ""))

            # 发送节点结束事件
            writer({
                "type": "step_end",
                "node": "valuation_analysis",
                "step_name": "💰 估值分析",
                "timestamp": datetime.now().isoformat()
            })

            return {
                "valuation_analysis": "\n".join(results),  # ✅ 拼接为单个字符串
                "valuation_analysis_json": json_result,    # ✅ 存储JSON数据
                "completed_steps": ["valuation_analysis"]
            }
        except Exception as e:
            self.logger.error(f"估值分析失败: {e}")
            writer = get_stream_writer()
            writer({
                "type": "error",
                "node": "valuation_analysis",
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            })
            return {
                "valuation_analysis": f"估值分析失败: {str(e)}",
                "valuation_analysis_json": None,
                "errors": [f"估值分析失败: {str(e)}"],
                "completed_steps": ["valuation_analysis"]
            }

    async def _capital_flow_analysis_node(self, state: StockAnalysisState) -> Dict[str, Any]:
        """资金分析节点 - 支持JSON格式输出"""
        try:
            self.logger.info(f"开始资金分析: {state['stock_code']}")

            # ✅ 获取stream writer
            writer = get_stream_writer()

            # 发送节点开始事件
            writer({
                "type": "step_start",
                "node": "capital_flow_analysis",
                "step_name": "💹 资金分析",
                "timestamp": datetime.now().isoformat()
            })

            results = []
            json_result = None

            # ✅ 根据输出格式调用智能体
            if state.get("output_format") == "json":
                # JSON格式输出 - 使用_generate_capital_signals方法
                result = await self.capital_agent._generate_capital_signals(
                    stock_code=state['stock_code'],
                    output_format="json"
                )
                # 解析JSON结果
                try:
                    json_result = json.loads(result) if isinstance(result, str) else result
                    # 同时保持文本格式（向后兼容）
                    text_result = json.dumps(json_result, ensure_ascii=False, indent=2)
                    results.append(text_result)
                except Exception as json_error:
                    self.logger.warning(f"JSON解析失败，使用文本格式: {json_error}")
                    results.append(result)
            else:
                # 原有文本处理逻辑
                async for event in self.capital_agent.stream_analyze(state['stock_code']):
                    event_type = event.get("type")

                    if event_type == "reasoning":
                        # ✅ 转发所有推理内容，不再过滤
                        content = event.get("content", "")
                        if content:
                            writer({
                                "type": "reasoning",
                                "node": "capital_flow_analysis",
                                "content": content,
                                "timestamp": datetime.now().isoformat()
                            })

                    elif event_type == "final_answer":
                        # 只保存结果，不发送到UI
                        results.append(event.get("content", ""))

            # 🆕 生成资金流向图表
            try:
                self.logger.info(f"开始生成资金流向图表: {state['stock_code']}")
                chart_result = await self.capital_agent.generate_capital_flow_charts(state['stock_code'])
                if not chart_result.get("error"):
                    self.logger.info(f"资金流向图表生成成功: {len(chart_result)}个图表")
                else:
                    self.logger.warning(f"资金流向图表生成失败: {chart_result.get('error')}")
            except Exception as chart_error:
                self.logger.error(f"生成资金流向图表异常: {chart_error}")

            # 发送节点结束事件
            writer({
                "type": "step_end",
                "node": "capital_flow_analysis",
                "step_name": "💹 资金分析",
                "timestamp": datetime.now().isoformat()
            })

            return {
                "capital_flow_analysis": "\n".join(results),  # ✅ 拼接为单个字符串
                "capital_flow_analysis_json": json_result,    # ✅ 存储JSON数据
                "completed_steps": ["capital_flow_analysis"]
            }
        except Exception as e:
            self.logger.error(f"资金分析失败: {e}")
            writer = get_stream_writer()
            writer({
                "type": "error",
                "node": "capital_flow_analysis",
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            })
            return {
                "capital_flow_analysis": f"资金分析失败: {str(e)}",
                "capital_flow_analysis_json": None,
                "errors": [f"资金分析失败: {str(e)}"],
                "completed_steps": ["capital_flow_analysis"]
            }

    async def _fundamental_analysis_node(self, state: StockAnalysisState) -> Dict[str, Any]:
        """财务分析节点 - 支持JSON格式输出"""
        try:
            self.logger.info(f"开始财务分析: {state['stock_code']}")

            # ✅ 获取stream writer
            writer = get_stream_writer()

            # 发送节点开始事件
            writer({
                "type": "step_start",
                "node": "fundamental_analysis",
                "step_name": "📊 财务分析",
                "timestamp": datetime.now().isoformat()
            })

            results = []
            json_result = None

            # ✅ 根据输出格式调用智能体
            if state.get("output_format") == "json":
                # JSON格式输出 - 使用_generate_fundamental_signals方法
                result = await self.fundamental_agent._generate_fundamental_signals(
                    stock_code=state['stock_code'],
                    output_format="json"
                )
                # 解析JSON结果
                try:
                    json_result = json.loads(result) if isinstance(result, str) else result
                    # 同时保持文本格式（向后兼容）
                    text_result = json.dumps(json_result, ensure_ascii=False, indent=2)
                    results.append(text_result)
                except Exception as json_error:
                    self.logger.warning(f"JSON解析失败，使用文本格式: {json_error}")
                    results.append(result)
            else:
                # 原有文本处理逻辑
                async for event in self.fundamental_agent.stream_analyze(state['stock_code']):
                    event_type = event.get("type")

                    if event_type == "reasoning":
                        # ✅ 转发所有推理内容，不再过滤
                        content = event.get("content", "")
                        if content:
                            writer({
                                "type": "reasoning",
                                "node": "fundamental_analysis",
                                "content": content,
                                "timestamp": datetime.now().isoformat()
                            })

                    elif event_type == "final_answer":
                        # 只保存结果，不发送到UI
                        results.append(event.get("content", ""))

            # 🆕 生成财务分析图表数据（只获取数据，不渲染）
            chart_data = {}
            try:
                self.logger.info(f"开始生成财务分析图表数据: {state['stock_code']}")
                chart_result = await self.fundamental_agent.generate_financial_charts(state['stock_code'])
                if not chart_result.get("error"):
                    chart_data = chart_result
                    self.logger.info(f"✅ 财务分析图表数据生成成功: {len(chart_result)}个图表")
                else:
                    self.logger.warning(f"⚠️ 财务分析图表数据生成失败: {chart_result.get('error')}")
            except Exception as chart_error:
                self.logger.error(f"❌ 生成财务分析图表数据异常: {chart_error}")

            # 发送节点结束事件
            writer({
                "type": "step_end",
                "node": "fundamental_analysis",
                "step_name": "📊 财务分析",
                "timestamp": datetime.now().isoformat()
            })

            return {
                "fundamental_analysis": "\n".join(results),  # ✅ 拼接为单个字符串
                "fundamental_analysis_json": json_result,    # ✅ 存储JSON数据
                "fundamental_charts": chart_data,            # 🆕 存储图表数据
                "completed_steps": ["fundamental_analysis"]
            }
        except Exception as e:
            self.logger.error(f"财务分析失败: {e}")
            writer = get_stream_writer()
            writer({
                "type": "error",
                "node": "fundamental_analysis",
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            })
            return {
                "fundamental_analysis": f"财务分析失败: {str(e)}",
                "fundamental_analysis_json": None,
                "fundamental_charts": None,
                "errors": [f"财务分析失败: {str(e)}"],
                "completed_steps": ["fundamental_analysis"]
            }

    async def _investment_summary_node(self, state: StockAnalysisState) -> Dict[str, Any]:
        """投资总结节点 - 支持JSON格式输出和维度数据聚合"""
        try:
            self.logger.info(f"开始投资总结: {state['stock_code']}")

            # ✅ 获取stream writer
            writer = get_stream_writer()

            # 发送节点开始事件
            writer({
                "type": "step_start",
                "node": "investment_summary",
                "step_name": "🎯 投资总结",
                "timestamp": datetime.now().isoformat()
            })

            # ✅ 获取实际执行的分析维度
            executed_dimensions = state.get("analysis_dimensions", [])
            if not executed_dimensions:
                executed_dimensions = ["technical", "news", "valuation", "fundamental", "capital"]
            
            # 添加详细的调试日志
            self.logger.info(f"🔍 _investment_summary_node: stock_code={state['stock_code']}, "
                           f"executed_dimensions={executed_dimensions}, "
                           f"output_format={state.get('output_format', 'text')}")
            
            # ✅ 关键修复：详细记录状态数据完整性
            self.logger.info(f"📊 状态数据完整性检查:")
            self.logger.info(f"  - technical_analysis: {state.get('technical_analysis') is not None}")
            self.logger.info(f"  - news_analysis: {state.get('news_analysis') is not None}")
            self.logger.info(f"  - valuation_analysis: {state.get('valuation_analysis') is not None}")
            self.logger.info(f"  - fundamental_analysis: {state.get('fundamental_analysis') is not None}")
            self.logger.info(f"  - capital_flow_analysis: {state.get('capital_flow_analysis') is not None}")
            self.logger.info(f"  - technical_analysis_json: {state.get('technical_analysis_json') is not None}")
            self.logger.info(f"  - news_analysis_json: {state.get('news_analysis_json') is not None}")
            self.logger.info(f"  - valuation_analysis_json: {state.get('valuation_analysis_json') is not None}")
            self.logger.info(f"  - fundamental_analysis_json: {state.get('fundamental_analysis_json') is not None}")
            self.logger.info(f"  - capital_flow_analysis_json: {state.get('capital_flow_analysis_json') is not None}")

            # 等待对应维度的分析完成
            dimension_to_step = {
                "technical": "technical_analysis",
                "news": "news_analysis",
                "valuation": "valuation_analysis",
                "fundamental": "fundamental_analysis",
                "capital": "capital_flow_analysis"
            }

            required_steps = [dimension_to_step[dim] for dim in executed_dimensions if dim in dimension_to_step]
            completed = state.get("completed_steps", [])

            if not all(step in completed for step in required_steps):
                self.logger.warning(f"部分分析未完成，已完成: {completed}，需要: {required_steps}")

            results = []
            json_result = None
            charts_result = None

            # ✅ 根据输出格式调用智能体
            if state.get("output_format") == "json":
                # 收集各维度JSON数据
                dimension_data = {
                    "technical_analysis_json": state.get("technical_analysis_json"),
                    "news_analysis_json": state.get("news_analysis_json"),
                    "valuation_analysis_json": state.get("valuation_analysis_json"),
                    "fundamental_analysis_json": state.get("fundamental_analysis_json"),
                    "capital_flow_analysis_json": state.get("capital_flow_analysis_json")
                }
                
                # JSON格式输出 - 调用投资总结智能体的JSON聚合方法
                result = await self.summary_agent._create_investment_report(
                    stock_code=state['stock_code'],
                    output_format="json",
                    dimension_data=dimension_data
                )
                
                # 解析JSON结果
                try:
                    json_result = json.loads(result) if isinstance(result, str) else result
                    # 同时保持文本格式（向后兼容）
                    text_result = json.dumps(json_result, ensure_ascii=False, indent=2)
                    results.append(text_result)
                    
                    # ✅ 关键修复：增强图表数据提取和验证逻辑
                    self.logger.info(f"🎯 投资总结智能体返回的JSON结果类型: {type(json_result)}")
                    self.logger.info(f"🎯 投资总结智能体返回的JSON结果键: {list(json_result.keys()) if isinstance(json_result, dict) else '不是字典'}")
                    
                    # 从JSON结果中提取图表数据
                    if isinstance(json_result, dict):
                        # 多层级查找图表数据
                        chart_data_sources = [
                            json_result.get("radar_chart_data"),
                            json_result.get("chart_data"),
                            json_result.get("charts"),
                            json_result.get("visualization_data"),
                            json_result.get("investment_charts")
                        ]
                        
                        charts_result = {}
                        for chart_source in chart_data_sources:
                            if chart_source and isinstance(chart_source, dict) and chart_source:
                                charts_result = chart_source
                                self.logger.info(f"✅ 成功从投资总结智能体提取图表数据: {list(chart_source.keys())}")
                                break
                        
                        # 验证图表数据完整性
                        if charts_result:
                            # 确保包含五维雷达图数据
                            if "five_dimension_radar" not in charts_result:
                                # 尝试从其他字段查找雷达图数据
                                radar_fields = ["radar_chart_data", "radar_data", "five_dimension"]
                                for field in radar_fields:
                                    if field in charts_result:
                                        radar_data = charts_result[field]
                                        if isinstance(radar_data, dict):
                                            charts_result["five_dimension_radar"] = radar_data
                                            self.logger.info(f"✅ 从字段 '{field}' 提取雷达图数据")
                                            break
                            
                            # 验证雷达图数据格式
                            if "five_dimension_radar" in charts_result:
                                radar_data = charts_result["five_dimension_radar"]
                                if isinstance(radar_data, dict):
                                    dimensions = radar_data.get("dimensions", [])
                                    scores = radar_data.get("scores", [])
                                    self.logger.info(f"✅ 五维雷达图数据验证: 维度={dimensions}, 评分={scores}")
                                    
                                    # 确保数据格式正确
                                    if not dimensions or not scores or len(dimensions) != len(scores):
                                        self.logger.warning("❌ 雷达图数据格式不正确，使用默认数据")
                                        charts_result["five_dimension_radar"] = {
                                            "dimensions": ["技术面", "消息面", "估值面", "财务面", "资金面"],
                                            "scores": [5.0, 5.0, 5.0, 5.0, 5.0],
                                            "max_score": 10,
                                            "average_score": 5.0
                                        }
                                else:
                                    self.logger.warning("❌ 雷达图数据不是字典类型，使用默认数据")
                                    charts_result["five_dimension_radar"] = {
                                        "dimensions": ["技术面", "消息面", "估值面", "财务面", "资金面"],
                                        "scores": [5.0, 5.0, 5.0, 5.0, 5.0],
                                        "max_score": 10,
                                        "average_score": 5.0
                                    }
                        else:
                            self.logger.warning("❌ 投资总结智能体返回的JSON中没有有效的图表数据字段")
                            charts_result = {}
                    else:
                        charts_result = {}
                        self.logger.warning("❌ 投资总结智能体返回的结果不是字典类型")
                        
                except Exception as json_error:
                    self.logger.error(f"❌ JSON解析失败: {json_error}")
                    results.append(result)
                    charts_result = {}
                    self.logger.error("❌ JSON解析失败，图表数据设置为空")
            else:
                # ✅ 原有文本处理逻辑 - 收集已执行维度的分析结果
                analysis_context = {
                    "stock_code": state['stock_code'],
                    "technical_analysis": state.get("technical_analysis", ""),
                    "news_analysis": state.get("news_analysis", ""),
                    "valuation_analysis": state.get("valuation_analysis", ""),
                    "fundamental_analysis": state.get("fundamental_analysis", ""),
                    "capital_flow_analysis": state.get("capital_flow_analysis", ""),
                    "analysis_dimensions": executed_dimensions  # 传递实际执行的维度
                }

                # ✅ 流式执行智能体，传递其他维度的分析结果作为上下文
                async for event in self.summary_agent.stream_analyze(state['stock_code']):
                    event_type = event.get("type")

                    if event_type == "reasoning":
                        # ✅ 转发所有推理内容，不再过滤
                        content = event.get("content", "")
                        if content:
                            writer({
                                "type": "reasoning",
                                "node": "investment_summary",
                                "content": content,
                                "timestamp": datetime.now().isoformat()
                            })

                    elif event_type == "final_answer":
                        # 只保存结果，不发送到UI
                        results.append(event.get("content", ""))
                
                # 文本模式下，优先从JSON数据提取评分，确保与文本中显示的评分一致
                try:
                    # ✅ 关键修复：优先从JSON数据中提取评分，而不是从文本中提取
                    dimension_scores = {}
                    
                    # JSON数据映射
                    json_data_map = {
                        "technical": state.get("technical_analysis_json"),
                        "news": state.get("news_analysis_json"),
                        "valuation": state.get("valuation_analysis_json"),
                        "fundamental": state.get("fundamental_analysis_json"),
                        "capital": state.get("capital_flow_analysis_json")
                    }
                    
                    # 文本数据映射（作为备份）
                    text_data_map = {
                        "technical": state.get("technical_analysis", ""),
                        "news": state.get("news_analysis", ""),
                        "valuation": state.get("valuation_analysis", ""),
                        "fundamental": state.get("fundamental_analysis", ""),
                        "capital": state.get("capital_flow_analysis", "")
                    }
                    
                    # 从JSON数据或文本中提取评分
                    for dim in ["technical", "news", "valuation", "fundamental", "capital"]:
                        json_data = json_data_map.get(dim)
                        
                        # 优先尝试从JSON数据提取评分
                        if json_data and isinstance(json_data, dict):
                            score = json_data.get("score")
                            if score is not None:
                                try:
                                    dimension_scores[dim] = float(score)
                                    self.logger.info(f"✅ 从{dim}维度JSON数据中提取评分: {score}")
                                    continue
                                except (ValueError, TypeError):
                                    pass
                        
                        # 如果JSON数据不存在或提取失败，从文本中提取
                        analysis_text = text_data_map.get(dim, "")
                        if analysis_text and analysis_text.strip():
                            score = self._extract_score_from_text(analysis_text, dim)
                            dimension_scores[dim] = score
                            self.logger.info(f"✅ 从{dim}维度文本中提取评分: {score}")
                        else:
                            dimension_scores[dim] = 5.0
                            self.logger.warning(f"⚠️ {dim}维度数据为空，使用默认评分5.0")
                    
                    # 生成雷达图数据 - 使用实际提取的评分
                    charts_result = {
                        "five_dimension_radar": {
                            "dimensions": ["技术面", "消息面", "估值面", "财务面", "资金面"],
                            "scores": [
                                dimension_scores.get("technical", 5.0),
                                dimension_scores.get("news", 5.0),
                                dimension_scores.get("valuation", 5.0),
                                dimension_scores.get("fundamental", 5.0),
                                dimension_scores.get("capital", 5.0)
                            ],
                            "max_score": 10,
                            "average_score": sum(dimension_scores.values()) / len(dimension_scores) if dimension_scores else 5.0
                        }
                    }
                    self.logger.info(f"✅ 文本模式下生成雷达图数据，实际评分: {dimension_scores}")
                except Exception as e:
                    charts_result = {}
                    self.logger.error(f"❌ 文本模式下图表数据生成失败: {e}")

            summary = "\n".join(results) if results else "投资总结生成失败"

            # 发送节点结束事件
            writer({
                "type": "step_end",
                "node": "investment_summary",
                "step_name": "🎯 投资总结",
                "timestamp": datetime.now().isoformat()
            })

            # ✅ 关键修复：确保charts_result不为None
            if charts_result is None:
                self.logger.warning("❌ charts_result为None，使用空字典作为默认值")
                charts_result = {}

            # ✅ 发送final_result事件，包含完整的分析状态（包含JSON数据和图表数据）
            complete_state = {
                "stock_code": state.get("stock_code", ""),
                "user_query": state.get("user_query", ""),
                "analysis_dimensions": executed_dimensions,  # 使用实际执行的维度
                "technical_analysis": state.get("technical_analysis", ""),
                "news_analysis": state.get("news_analysis", ""),
                "valuation_analysis": state.get("valuation_analysis", ""),
                "fundamental_analysis": state.get("fundamental_analysis", ""),
                "capital_flow_analysis": state.get("capital_flow_analysis", ""),
                "investment_summary": summary,
                "output_format": state.get("output_format", "text"),
                "technical_analysis_json": state.get("technical_analysis_json"),
                "news_analysis_json": state.get("news_analysis_json"),
                "valuation_analysis_json": state.get("valuation_analysis_json"),
                "fundamental_analysis_json": state.get("fundamental_analysis_json"),
                "capital_flow_analysis_json": state.get("capital_flow_analysis_json"),
                "investment_summary_json": json_result,
                "investment_charts": charts_result,  # ✅ 新增：图表数据
                "technical_charts": state.get("technical_charts"),  # 🆕 添加技术分析图表数据
                "fundamental_charts": state.get("fundamental_charts"),  # 🆕 添加财务分析图表数据
                "errors": state.get("errors", []),
                "completed_steps": state.get("completed_steps", []) + ["investment_summary"]
            }

            # 详细记录图表数据状态
            if charts_result and isinstance(charts_result, dict):
                self.logger.info(f"发送final_result事件，包含图表数据: {list(charts_result.keys())}")
                if "five_dimension_radar" in charts_result:
                    radar_data = charts_result["five_dimension_radar"]
                    if isinstance(radar_data, dict):
                        self.logger.info(f"五维雷达图数据: 维度={radar_data.get('dimensions')}, 评分={radar_data.get('scores')}")
            else:
                self.logger.warning("图表数据为空或无效")

            writer({
                "type": "final_result",
                "state": complete_state,
                "timestamp": datetime.now().isoformat()
            })

            return {
                "investment_summary": summary,
                "investment_summary_json": json_result,  # ✅ 存储JSON数据
                "investment_charts": charts_result,      # ✅ 新增：图表数据
                "completed_steps": ["investment_summary"]
            }
        except Exception as e:
            self.logger.error(f"投资总结失败: {e}")
            writer = get_stream_writer()
            writer({
                "type": "error",
                "node": "investment_summary",
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            })
            return {
                "investment_summary": f"投资总结失败: {str(e)}",
                "investment_summary_json": None,
                "investment_charts": None,  # ✅ 新增：图表数据（失败时为None）
                "errors": [f"投资总结失败: {str(e)}"],
                "completed_steps": ["investment_summary"]
            }


    async def execute(self, stock_code: str, user_query: str = "",
                      thread_id: Optional[str] = None,
                      analysis_dimensions: Optional[List[str]] = None,
                      output_format: Literal["text", "json"] = "text") -> Dict[str, Any]:
        """
        执行股票分析工作流

        Args:
            stock_code: 股票代码
            user_query: 用户查询
            thread_id: 线程ID
            analysis_dimensions: 分析维度列表（可选，支持部分维度分析）
            output_format: 输出格式，支持 "text" 或 "json"

        Returns:
            分析结果字典
        """
        initial_state: StockAnalysisState = {
            "stock_code": stock_code,
            "user_query": user_query,
            "analysis_dimensions": analysis_dimensions or [],  # ✅ 支持指定维度
            "technical_analysis": "",
            "news_analysis": "",
            "valuation_analysis": "",
            "fundamental_analysis": "",
            "capital_flow_analysis": "",
            "investment_summary": "",
            "output_format": output_format,  # ✅ 使用传入的输出格式
            "technical_analysis_json": None,
            "news_analysis_json": None,
            "valuation_analysis_json": None,
            "fundamental_analysis_json": None,
            "capital_flow_analysis_json": None,
            "investment_summary_json": None,
            "technical_charts": None,
            "fundamental_charts": None,
            "investment_charts": None,
            "errors": [],
            "completed_steps": []
        }

        try:
            # 执行工作流
            final_state = await self.app.ainvoke(initial_state)

            # 根据实际执行的维度返回结果
            executed_dimensions = final_state.get("analysis_dimensions", [])
            if not executed_dimensions:
                executed_dimensions = ["technical", "news", "valuation", "fundamental", "capital"]

            # 只返回实际执行的分析结果 - 修复维度键名不一致问题
            analysis_results = {"summary": final_state.get("investment_summary")}
            for dim in executed_dimensions:
                if dim == "technical":
                    analysis_results["technical"] = final_state.get("technical_analysis")
                elif dim == "news":
                    analysis_results["news"] = final_state.get("news_analysis")
                elif dim == "valuation":
                    analysis_results["valuation"] = final_state.get("valuation_analysis")
                elif dim == "fundamental":
                    analysis_results["fundamental"] = final_state.get("fundamental_analysis")
                elif dim == "capital":
                    analysis_results["capital"] = final_state.get("capital_flow_analysis")  # 修复：使用"capital"而不是"capital_flow"

            return {
                "stock_code": stock_code,
                "user_query": user_query,
                "analysis_results": analysis_results,
                "workflow_status": final_state.get("completed_steps", []),
                "errors": final_state.get("errors", []),
                "success": len(final_state.get("errors", [])) == 0
            }

        except Exception as e:
            return {
                "stock_code": stock_code,
                "user_query": user_query,
                "error": str(e),
                "success": False
            }

    async def stream_execute(self, stock_code: str, user_query: str = "",
                              thread_id: Optional[str] = None,
                              analysis_dimensions: Optional[List[str]] = None,
                              output_format: Literal["text", "json"] = "text") -> AsyncIterator[Dict[str, Any]]:
        """
        流式执行股票分析工作流（使用LangGraph自定义事件流）

        ✅ 参考债券分析实现：
        - 使用StateGraph定义的并行边（真正的LangGraph并行）
        - 使用get_stream_writer()在节点内发送自定义事件
        - 使用stream_mode="custom"接收实时推理内容

        Args:
            stock_code: 股票代码
            user_query: 用户查询
            thread_id: 线程ID
            analysis_dimensions: 分析维度列表（可选，支持部分维度分析）
            output_format: 输出格式，支持 "text" 或 "json"

        Yields:
            工作流执行的每个步骤事件，包括智能体的实时推理内容
        """
        initial_state: StockAnalysisState = {
            "stock_code": stock_code,
            "user_query": user_query,
            "analysis_dimensions": analysis_dimensions or [],  # ✅ 支持指定维度（空列表=全量分析）
            "technical_analysis": "",
            "news_analysis": "",
            "valuation_analysis": "",
            "fundamental_analysis": "",
            "capital_flow_analysis": "",
            "investment_summary": "",
            "output_format": output_format,  # ✅ 使用传入的输出格式
            "technical_analysis_json": None,
            "news_analysis_json": None,
            "valuation_analysis_json": None,
            "fundamental_analysis_json": None,
            "capital_flow_analysis_json": None,
            "investment_summary_json": None,
            "technical_charts": None,
            "fundamental_charts": None,
            "investment_charts": None,
            "errors": [],
            "completed_steps": []
        }

        try:
            # ✅ 使用stream_mode="custom"接收节点内通过writer发送的自定义事件
            final_state = None
            async for event in self.app.astream(
                initial_state,
                stream_mode="custom"  # 关键：只接收自定义事件
            ):
                # 直接转发自定义事件给UI层
                yield event

                # ✅ 捕获final_result事件中的状态
                if event.get("type") == "final_result":
                    final_state = event.get("state", {})

            # ✅ 如果没有从事件中获取到状态，发送完成事件
            if not final_state:
                # 由于没有checkpointer，我们通过initial_state构建一个基本的最终状态
                # 注意：实际结果已通过final_result事件发送
                yield {
                    "type": "workflow_complete",
                    "state": initial_state,  # 基本状态
                    "timestamp": datetime.now().isoformat()
                }

        except Exception as e:
            self.logger.error(f"流式执行失败: {e}", exc_info=True)
            yield {
                "type": "error",
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }


    def _extract_score_from_text(self, analysis_text: str, dimension: str) -> float:
        """
        从文本分析结果中智能提取评分
        
        Args:
            analysis_text: 分析文本内容
            dimension: 维度名称
            
        Returns:
            提取的评分 (0-10)
        """
        try:
            # 查找评分相关的关键词和模式
            score_patterns = [
                (r'评分[：:]\s*(\d+(?:\.\d+)?)', 1),  # "评分：8.5"
                (r'得分[：:]\s*(\d+(?:\.\d+)?)', 1),  # "得分：7.2"
                (r'score[：:]\s*(\d+(?:\.\d+)?)', 1),  # "score: 6.8"
                (r'(\d+(?:\.\d+)?)\s*分', 1),  # "8.5分"
                (r'(\d+(?:\.\d+)?)\s*\/\s*10', 1),  # "7.2/10"
                (r'(\d+(?:\.\d+)?)\s*\/\s*10', 1),  # "6.8 / 10"
            ]
            
            for pattern, group in score_patterns:
                import re
                match = re.search(pattern, analysis_text, re.IGNORECASE)
                if match:
                    score = float(match.group(group))
                    # 确保评分在合理范围内
                    if 0 <= score <= 10:
                        self.logger.info(f"✅ 从{dimension}维度文本中提取到评分: {score}")
                        return score
                    else:
                        self.logger.warning(f"⚠️ 提取的评分超出范围: {score}，使用默认值5.0")
                        return 5.0
            
            # 如果找不到明确的评分，根据文本情感分析估算
            positive_keywords = ['看好', '推荐', '买入', '强势', '上涨', '突破', '良好', '优秀', '积极']
            negative_keywords = ['谨慎', '观望', '卖出', '弱势', '下跌', '破位', '一般', '较差', '消极']
            
            positive_count = sum(1 for keyword in positive_keywords if keyword in analysis_text)
            negative_count = sum(1 for keyword in negative_keywords if keyword in analysis_text)
            
            if positive_count > negative_count:
                score = 7.5  # 积极评价
            elif negative_count > positive_count:
                score = 2.5  # 消极评价
            else:
                score = 5.0  # 中性评价
                
            self.logger.info(f"🔍 基于文本情感分析，{dimension}维度估算评分: {score}")
            return score
            
        except Exception as e:
            self.logger.warning(f"❌ 从{dimension}维度文本提取评分失败: {e}，使用默认值5.0")
            return 5.0

# 创建全局实例
stock_analysis_workflow = StockAnalysisWorkflow()


async def analyze_stock(stock_code: str, user_query: str = "",
                       analysis_dimensions: Optional[List[str]] = None,
                       output_format: Literal["text", "json"] = "text") -> Dict[str, Any]:
    """
    分析股票的便捷函数

    Args:
        stock_code: 股票代码
        user_query: 用户查询
        analysis_dimensions: 分析维度列表（可选，支持部分维度分析）
                           ["technical", "news", "valuation", "fundamental", "capital"]
        output_format: 输出格式，支持 "text" 或 "json"

    Returns:
        分析结果
    """
    return await stock_analysis_workflow.execute(
        stock_code, user_query,
        analysis_dimensions=analysis_dimensions,
        output_format=output_format
    )